home *** CD-ROM | disk | FTP | other *** search
/ Windows Expert / Windows Expert.iso / windownt / queue.zip / QUEUE / QUEUE.C < prev    next >
C/C++ Source or Header  |  1992-10-11  |  16KB  |  668 lines
  1. /************************************************************************\
  2. * The enclosed files, "the software," is provided by 
  3. * Microsoft Corporation "as is" without warranty of any kind. 
  4. * MICROSOFT DISCLAIMS ALL WARRANTIES, EITHER EXPRESS OR IMPLIED, 
  5. * INCLUDING BUT NOT LIMITED TO IMPLIED WARRANTIES OF MERCHANTABILITY 
  6. * AND FITNESS FOR A PARTICULAR PURPOSE.  You assume all risks of 
  7. * using the software.
  8. * 
  9. * The software is Copyright (c) 1992 Microsoft Corporation.
  10. * Original Author: John M. Hall, Microsoft SDE  9/1/92
  11. *
  12. * You are granted the right to freely distribute this software.
  13. * You are granted the right to make changes provided this comment block
  14. * is retained without modification and you acknowledge the changes.
  15. * 
  16. \************************************************************************/
  17. /************************************************************************\
  18. *
  19. *  MODULE:      QUEUE.C
  20. *
  21. *  PURPOSE:     Simple shared memory management.
  22. *
  23. *  FUNCTIONS:   QueueEntry() - DLL entry point
  24. *
  25. *
  26. *
  27. *  COMMENTS:    
  28. *
  29. ************************************************************************/
  30. #define QUEUE_DLL_C
  31.  
  32. #include <stdio.h>
  33. #include <memory.h>
  34. #include <windows.h>
  35. #include <winbase.h>
  36. #include <string.h>
  37. #include <malloc.h>
  38.  
  39. #define DEBUG
  40. #define DOSWIN32
  41. #include "Queue.h"
  42.  
  43. /************************************************************************\
  44. *
  45. *  FUNCTION:    QueueEntry
  46. *
  47. *  INPUTS:      hDLL       - handle of DLL
  48. *               dwReason   - indicates why DLL called
  49. *               lpReserved - reserved
  50. *
  51. *  RETURNS:     TRUE for success, FALSE on error
  52. *
  53. *               Note that the retuRn value is used only when
  54. *               dwReason = DLL_PROCESS_ATTACH.
  55. *
  56. *  GLOBAL VARS: See Queue.h for list of global variables.
  57. *
  58. *  COMMENTS:    Modified from simple dll sample
  59. *
  60. \************************************************************************/
  61. void queue_cleanup()
  62. {
  63.     return;
  64. }
  65.  
  66. BOOL  QueueEntry (HANDLE hDLL, DWORD dwReason, LPVOID lpReserved)
  67. { 
  68.   switch (dwReason)
  69.   { 
  70.   case DLL_PROCESS_ATTACH:
  71.       { 
  72.       /******************************************************************\
  73.       *  DLL is attaching to the address space of the current process.
  74.       \******************************************************************/
  75.       BOOL bFlag;
  76.  
  77.       fp_log = fopen( "hp_shr.log", "a+");
  78.  
  79.       //
  80.       // Attach control memory for Queues
  81.       //
  82.       bFlag = attach_ctl();
  83.  
  84.       if (!bFlag)
  85.           fclose(fp_log);
  86.  
  87.       AssertBox( bFlag, "Attach Memory");
  88.  
  89.       return(bFlag);
  90.       break;
  91.     }
  92.  
  93.     case DLL_THREAD_ATTACH:
  94.  
  95.       /******************************************************************\
  96.       *  A new thread is being created in the current process.
  97.       \******************************************************************/
  98.  
  99.       break;
  100.     case DLL_THREAD_DETACH:
  101.  
  102.       /******************************************************************\
  103.       *  A thread is exiting cleanly.
  104.       \******************************************************************/
  105.  
  106.       break;
  107.     case DLL_PROCESS_DETACH:
  108.  
  109.       /******************************************************************\
  110.       *  The calling process is detaching the DLL from its address space.
  111.       \******************************************************************/
  112.  
  113.       queue_cleanup();
  114.       fclose(fp_log);
  115.       UnmapViewOfFile(lpCtrlBase);
  116.       CloseHandle(hControl);
  117.       break;
  118.   }
  119.   return TRUE;
  120.   UNREFERENCED_PARAMETER(hDLL);
  121.   UNREFERENCED_PARAMETER(lpReserved);
  122. }
  123.  
  124. VOID MyAssertBox( const char *title, const char *exp,
  125.         const char *file, int line)
  126.         {
  127.         char buffer[512];
  128.  
  129.         sprintf( buffer, "Assert(%s) at (%s::%d)", exp, file, line);
  130.         MessageBox (NULL, title, buffer, MB_OK | MB_ICONHAND);
  131.         }
  132.  
  133. BOOL attach_ctl()
  134.     {
  135.     PQUEUE_DATA  pqd;
  136.     int ii;
  137.  
  138.  
  139.     SetErrorMode(SEM_FAILCRITICALERRORS);
  140.  
  141.     hGlobalSem = CreateSemaphore( NULL, 0, 1, "JMH_QUEUE_SEM");
  142.  
  143.     if (hGlobalSem)
  144.         {
  145.         //
  146.         // This is correct, we use GetLastError() to find out how the
  147.         // function succeeded!
  148.         //
  149.         if (GetLastError() == ERROR_ALREADY_EXISTS)
  150.             {
  151.             //
  152.             // We didn't create it, just attach to it so we need to
  153.             // wait till its free.
  154.             //
  155.             WaitForSingleObject(hGlobalSem, INFINITE);
  156.             dwProcesses++;
  157.             }
  158.         else
  159.             {
  160.             dwProcesses = 1;
  161.             }
  162.  
  163.         //
  164.         // The heart of the matter, create the memory mapped files.
  165.         //
  166.         hControl = CreateFileMapping(
  167.             (HANDLE) INFINITE,
  168.             NULL,
  169.             PAGE_READWRITE,
  170.             0,
  171.             64 * 1024,
  172.             "JMH_QUEUE_CTRL");
  173.  
  174.         if (hControl == NULL)
  175.             fprintf( fp_log,  "CreateFileMapping Failed\n");
  176.  
  177.         //
  178.         // Now attach the memory to the correct addresses
  179.         //
  180.         lpCtrlBase = MapViewOfFile( hControl, FILE_MAP_WRITE, 0, 0, 0);
  181.  
  182.         if (lpCtrlBase == NULL)
  183.             {
  184.             fprintf( fp_log,  "MapViewOfFile Failed\n");
  185.             ReleaseSemaphore(hGlobalSem, 1, NULL);
  186.             CloseHandle(hControl);
  187.             return(FALSE);
  188.             }
  189.  
  190.         //
  191.         // First time initialization.
  192.         //
  193.         if (dwProcesses == 1)
  194.             {
  195.             iQueues = 0;
  196.             pqd = (PQUEUE_DATA) lpCtrlBase;
  197.  
  198.             for (ii = 0; ii < MAX_QUEUES; ii++ )
  199.                 {
  200.                 pqd[ii].aQueueName = 0;
  201.                 pqd[ii].sElements  = 0;
  202.                 }
  203.             }
  204.  
  205.         ReleaseSemaphore(hGlobalSem, 1, NULL);
  206.  
  207.         return(TRUE);
  208.  
  209.         }
  210.     else
  211.         return (FALSE);
  212.     }
  213.  
  214. BOOL IsValidHqueue( HQUEUE hq)
  215. {
  216.  
  217.     if (IsBadReadPtr(hq, sizeof(HQUEUE)) == TRUE)
  218.         return(FALSE);
  219.  
  220.     if (hq->dwMagic != MAGIC)
  221.         return(FALSE);
  222.  
  223.     return(TRUE);
  224. }
  225.  
  226. DWORD CreateQueue( PHQUEUE phq, int fQueueOrder, LPTSTR pszName)
  227. {
  228.     HQUEUE hq;
  229.     LPTSTR pszBase;
  230.     int    cb;
  231.     HANDLE hEvent;
  232.     HANDLE hSem;
  233.     int    iIndex = -1;
  234.     PQUEUE_DATA pqd;
  235.     int    ii;
  236.     ATOM   atom;
  237.  
  238.  
  239.     if (strncmp(pszName, "\\queues\\", 8) == 0)
  240.         pszName += 8;
  241.  
  242.     if ((cb = strlen(pszName)) == 0)
  243.         {
  244.         return(QUE_INVALID_NAME);
  245.         }
  246.  
  247.     if (strchr(pszName, '\\') != NULL)
  248.         {
  249.         return(QUE_INVALID_NAME);
  250.         }
  251.  
  252.     if ((atom = GlobalAddAtom(pszName)) == 0)
  253.         {
  254.         return(GetLastError());
  255.         }
  256.  
  257.     WaitForSingleObject(hGlobalSem, INFINITE);
  258.  
  259.     if (iQueues == QUEUE_LIMIT)
  260.         {
  261.         ReleaseSemaphore(hGlobalSem, 1, NULL);
  262.         GlobalDeleteAtom(atom);
  263.         return( QUE_NO_MEMORY);
  264.         }
  265.  
  266.     pqd = (PQUEUE_DATA) lpCtrlBase;
  267.  
  268.     ii = 0;
  269.     while (iIndex == -1 && ii < QUEUE_LIMIT)
  270.         {
  271.         if (pqd[ii].aQueueName == 0)
  272.             iIndex = ii;
  273.         else
  274.             ii++;
  275.         }
  276.  
  277.     if (iIndex == -1)
  278.         {
  279.         GlobalDeleteAtom(atom);
  280.         ReleaseSemaphore(hGlobalSem, 1, NULL);
  281.         return( QUE_NO_MEMORY);
  282.         }
  283.  
  284.  
  285.     pszBase = malloc(cb + 6);
  286.     strcpy( pszBase, pszName);
  287.     strcat( pszBase, "_Event");
  288.  
  289.     hEvent = CreateEvent( NULL, TRUE, FALSE, pszBase);
  290.  
  291.     if (hEvent == NULL)
  292.         {
  293.         ReleaseSemaphore(hGlobalSem, 1, NULL);
  294.         GlobalDeleteAtom(atom);
  295.         free(pszBase);
  296.         return(GetLastError());
  297.         }
  298.  
  299.     strcpy( pszBase, pszName);
  300.     strcat( pszBase, "_Sem");
  301.  
  302.     hSem = CreateSemaphore( NULL, 1, 1, pszBase);
  303.  
  304.     if (hSem == NULL)
  305.         {
  306.         ReleaseSemaphore(hGlobalSem, 1, NULL);
  307.         GlobalDeleteAtom(atom);
  308.         free(pszBase);
  309.         CloseHandle(hEvent);
  310.         return(GetLastError());
  311.         }
  312.  
  313.  
  314.     hq = (HQUEUE) malloc(sizeof(QUEUE));
  315.  
  316.     if (!hq)
  317.         {
  318.         ReleaseSemaphore(hGlobalSem, 1, NULL);
  319.         return( QUE_NO_MEMORY);
  320.         }
  321.  
  322.     hq->dwMagic   = MAGIC;
  323.     hq->hEvent    = hEvent;
  324.     hq->hSem      = hSem;
  325.     hq->iIndex    = iIndex;
  326.     hq->dwProcId  = GetCurrentProcessId();
  327.  
  328.     pqd[iIndex].wType      = fQueueOrder;
  329.     pqd[iIndex].dwCreator  = hq->dwProcId;
  330.     pqd[iIndex].aQueueName = atom;
  331.     pqd[iIndex].sElements  = 0;
  332.  
  333.     *phq = hq;
  334.     ReleaseSemaphore(hGlobalSem, 1, NULL);
  335.     return(0);
  336. }
  337.  
  338. DWORD CloseQueue( HQUEUE hq)
  339. {
  340.     PQUEUE_DATA pqd;
  341.     int         iIndex;
  342.     int         ii;
  343.     DWORD       dwProcId;
  344.     HANDLE      hSem;
  345.  
  346.     if (!IsValidHqueue( hq))
  347.         return(QUE_INVALID_HANDLE);
  348.  
  349.  
  350.     WaitForSingleObject(hSem = hq->hSem, INFINITE);
  351.  
  352.     iIndex = hq->iIndex;
  353.  
  354.     pqd = (PQUEUE_DATA) lpCtrlBase;
  355.  
  356.     dwProcId = hq->dwProcId;
  357.     CloseHandle(hq->hEvent);
  358.     free(hq);
  359.  
  360.     if (pqd[iIndex].aQueueName != 0)
  361.         GlobalDeleteAtom(pqd[iIndex].aQueueName);
  362.  
  363.     if (dwProcId != pqd[iIndex].dwCreator)
  364.         {
  365.         ReleaseSemaphore(hSem, 1, NULL);
  366.         CloseHandle(hSem);
  367.         return(0);
  368.         }
  369.  
  370.     for (ii = 0; ii < pqd[iIndex].sElements ; ii++)
  371.         {
  372.         ShrFree(pqd[iIndex].elem[ii].dwShrHandle);
  373.         }
  374.  
  375.     pqd[iIndex].aQueueName = 0;
  376.     pqd[iIndex].dwCreator  = 0;
  377.     ReleaseSemaphore(hSem, 1, NULL);
  378.     CloseHandle(hSem);
  379.     return(0);
  380. }
  381.  
  382. DWORD OpenQueue(LPDWORD ppid, PHQUEUE phq, LPTSTR pszName)
  383. {
  384.     HQUEUE hq;
  385.     LPTSTR pszBase;
  386.     int    cb;
  387.     HANDLE hEvent;
  388.     HANDLE hSem;
  389.     int    iIndex = -1;
  390.     PQUEUE_DATA pqd;
  391.     int    ii;
  392.     ATOM   atom;
  393.  
  394.  
  395.     pqd = (PQUEUE_DATA) lpCtrlBase;
  396.  
  397.     if (strncmp(pszName, "\\queues\\", 8) == 0)
  398.         pszName += 8;
  399.  
  400.     if ((cb = strlen(pszName)) == 0)
  401.         return(QUE_INVALID_NAME);
  402.  
  403.     if (strchr(pszName, '\\') != NULL)
  404.         return(QUE_INVALID_NAME);
  405.  
  406.     if ((atom = GlobalAddAtom(pszName)) == 0)
  407.         return(GetLastError());
  408.  
  409.     hq = (HQUEUE) malloc(sizeof(QUEUE));
  410.  
  411.     if (!hq)
  412.         return(QUE_NO_MEMORY);
  413.  
  414.     WaitForSingleObject(hGlobalSem, INFINITE);
  415.  
  416.     ii = 0;
  417.     while (iIndex == -1 && ii < QUEUE_LIMIT)
  418.         {
  419.         if (pqd[ii].aQueueName == atom)
  420.             iIndex = ii;
  421.         else
  422.             ii++;
  423.         }
  424.  
  425.     if (iIndex == -1)
  426.         {
  427.         free(hq);
  428.         ReleaseSemaphore(hGlobalSem, 1, NULL);
  429.         return( QUE_NAME_NOT_EXIST);
  430.         }
  431.  
  432.     pszBase = malloc(cb + 6);
  433.     strcpy( pszBase, pszName);
  434.     strcat( pszBase, "_Event");
  435.  
  436.     hEvent = CreateEvent( NULL, TRUE, FALSE, pszBase);
  437.  
  438.     if (hEvent == NULL)
  439.         {
  440.         free(hq);
  441.         GlobalDeleteAtom(atom);
  442.         free(pszBase);
  443.         return(GetLastError());
  444.         }
  445.  
  446.     strcpy( pszBase, pszName);
  447.     strcat( pszBase, "_Sem");
  448.  
  449.     hSem = CreateSemaphore( NULL, 1, 1, pszBase);
  450.  
  451.     if (hSem == NULL)
  452.         {
  453.         free(hq);
  454.         GlobalDeleteAtom(atom);
  455.         free(pszBase);
  456.         CloseHandle(hEvent);
  457.         return(GetLastError());
  458.         }
  459.  
  460.     hq->dwMagic   = MAGIC;
  461.     hq->hEvent    = hEvent;
  462.     hq->hSem      = hSem;
  463.     hq->iIndex    = iIndex;
  464.     hq->dwProcId  = GetCurrentProcessId();
  465.  
  466.     *ppid = pqd[iIndex].dwCreator;
  467.     *phq = hq;
  468.     ReleaseSemaphore(hGlobalSem, 1, NULL);
  469.     return(0);
  470. }
  471.  
  472. void push_down( int iIndex, int jj)
  473. {
  474.     PQUEUE_DATA pqd;
  475.     int ii;
  476.  
  477.     pqd = (PQUEUE_DATA) lpCtrlBase;
  478.  
  479.     for (ii = pqd[iIndex].sElements; ii > jj; ii--)
  480.         pqd[iIndex].elem[ii] = pqd[iIndex].elem[ii -1];
  481. }
  482.  
  483. void push_up( int iIndex, int jj)
  484. {
  485.     PQUEUE_DATA pqd;
  486.     int ii;
  487.  
  488.     pqd = (PQUEUE_DATA) lpCtrlBase;
  489.  
  490.     for (ii = jj; ii < pqd[iIndex].sElements; ii++)
  491.         pqd[iIndex].elem[ii] = pqd[iIndex].elem[ii +1];
  492. }
  493.  
  494. DWORD WriteQueue(HQUEUE hq, DWORD dwEventCode, DWORD dwShrHandle, DWORD dwPriority)
  495. {
  496.     int iIndex;
  497.     PQUEUE_DATA pqd;
  498.     int ii;
  499.  
  500.     if (!IsValidHqueue( hq))
  501.         return(QUE_INVALID_HANDLE);
  502.  
  503.     WaitForSingleObject( hq->hSem, INFINITE);
  504.  
  505.     iIndex = hq->iIndex;
  506.  
  507.     pqd = (PQUEUE_DATA) lpCtrlBase;
  508.  
  509.     if (pqd[iIndex].sElements == QUEUE_LIMIT)
  510.         {
  511.         ReleaseSemaphore(hq->hSem, 1, NULL);
  512.         return(QUE_NO_MEMORY);
  513.         }
  514.  
  515.     if (pqd[iIndex].sElements == 0)
  516.         ii = 0;
  517.     else
  518.         {
  519.         if (pqd[iIndex].wType == QUE_FIFO)
  520.             ii = pqd[iIndex].sElements;
  521.         if (pqd[iIndex].wType == QUE_LIFO)
  522.             {
  523.             push_down(iIndex, 0);
  524.             ii = 0;
  525.             }
  526.         if (pqd[iIndex].wType == QUE_PRIORITY)
  527.             {
  528.             ii = 0;
  529.             while (ii < pqd[iIndex].sElements &&
  530.                         dwPriority < pqd[iIndex].elem[ii].dwPriority)
  531.                 ii++;
  532.             if (ii != pqd[iIndex].sElements)
  533.                 push_down( iIndex, ii);
  534.             }
  535.         }
  536.  
  537.     pqd[iIndex].elem[ii].dwWriterId  = hq->dwProcId;
  538.     pqd[iIndex].elem[ii].dwShrHandle = dwShrHandle;
  539.     pqd[iIndex].elem[ii].dwEventCode = dwEventCode;
  540.     pqd[iIndex].elem[ii].dwPriority  = dwPriority;
  541.     pqd[iIndex].sElements++;
  542.  
  543.     SetEvent(hq->hEvent);
  544.     ReleaseSemaphore(hq->hSem, 1, NULL);
  545.     return(0);
  546.  
  547. }
  548.  
  549.                                   
  550. DWORD ReadQueue(HQUEUE hq, PQ_ELEMENT pqe, int iElement, BOOL fWait)
  551. {
  552.  
  553.     int iIndex;
  554.     PQUEUE_DATA pqd;
  555.  
  556.  
  557.     if (!IsValidHqueue( hq))
  558.         return(QUE_INVALID_HANDLE);
  559.  
  560.     WaitForSingleObject(hq->hSem, INFINITE);
  561.  
  562.     iIndex = hq->iIndex;
  563.  
  564.     pqd = (PQUEUE_DATA) lpCtrlBase;
  565.  
  566.     if (hq->dwProcId != pqd[iIndex].dwCreator)
  567.         {
  568.         ReleaseSemaphore(hq->hSem, 1, NULL);
  569.         return(QUE_INVALID_HANDLE);
  570.         }
  571.  
  572.     if (iElement < 0)
  573.         iElement = 0;
  574.  
  575.     if (iElement >= pqd[iIndex].sElements && iElement != 0)
  576.         {
  577.         ReleaseSemaphore(hq->hSem, 1, NULL);
  578.         return(QUE_ELEMENT_NOT_EXIST);
  579.         }
  580.  
  581.     if (pqd[iIndex].sElements == 0)
  582.         if (fWait)
  583.             {
  584.             while (pqd[iIndex].sElements == 0)
  585.                 {
  586.                 ReleaseSemaphore(hq->hSem, 1, NULL);
  587.                 WaitForSingleObject(hq->hEvent, INFINITE);
  588.                 WaitForSingleObject(hq->hSem, INFINITE);
  589.                 }
  590.             }
  591.         else
  592.             return(QUE_EMPTY);
  593.  
  594.     *pqe = pqd[iIndex].elem[iElement];
  595.     push_up(iIndex, iElement);
  596.     pqd[iIndex].sElements --;
  597.  
  598.     if (pqd[iIndex].sElements == 0)
  599.         ResetEvent(hq->hEvent);
  600.  
  601.     ReleaseSemaphore(hq->hSem, 1, NULL);
  602.     return(0);
  603.  
  604. }
  605.  
  606. HANDLE GetQueueEventHandle( HQUEUE hq)
  607. {
  608.     if (hq->dwMagic != MAGIC)
  609.         return(0);
  610.  
  611.     return( hq->hEvent);
  612. }
  613.  
  614.  
  615. DWORD PeekQueue(HQUEUE hq, PQ_ELEMENT pqe, int *piElement, BOOL fWait)
  616. {
  617.  
  618.     int iIndex;
  619.     PQUEUE_DATA pqd;
  620.  
  621.     if (!IsValidHqueue( hq))
  622.         return(QUE_INVALID_HANDLE);
  623.  
  624.     WaitForSingleObject(hq->hSem, INFINITE);
  625.  
  626.     iIndex = hq->iIndex;
  627.  
  628.     pqd = (PQUEUE_DATA) lpCtrlBase;
  629.  
  630.     if (hq->dwProcId != pqd[iIndex].dwCreator)
  631.         {
  632.         ReleaseSemaphore(hq->hSem, 1, NULL);
  633.         return(QUE_INVALID_HANDLE);
  634.         }
  635.  
  636.     if (*piElement < 0)
  637.         *piElement = 0;
  638.     else
  639.         (*piElement)++;
  640.  
  641.     if (*piElement >= pqd[iIndex].sElements && *piElement != 0)
  642.         {
  643.         ReleaseSemaphore(hq->hSem, 1, NULL);
  644.         return(QUE_ELEMENT_NOT_EXIST);
  645.         }
  646.  
  647.     if (pqd[iIndex].sElements == 0)
  648.         if (fWait)
  649.             {
  650.             while (pqd[iIndex].sElements == 0)
  651.                 {
  652.                 ReleaseSemaphore(hq->hSem, 1, NULL);
  653.                 WaitForSingleObject(hq->hEvent, INFINITE);
  654.                 WaitForSingleObject(hq->hSem, INFINITE);
  655.                 }
  656.             }
  657.         else
  658.             return(QUE_EMPTY);
  659.  
  660.     *pqe = pqd[iIndex].elem[*piElement];
  661.  
  662.     ReleaseSemaphore(hq->hSem, 1, NULL);
  663.     return(0);
  664.  
  665.  
  666. }
  667.  
  668.